package com.sztx.pay.center.dataaccess.dao.hbase.impl;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.codehaus.plexus.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.hadoop.hbase.TableCallback;
import org.springframework.stereotype.Repository;

import com.alibaba.fastjson.JSONObject;
import com.sztx.pay.center.common.util.DateUtil;
import com.sztx.pay.center.common.util.HbaseUtils;
import com.sztx.pay.center.common.util.StringUtil;
import com.sztx.pay.center.dataaccess.dao.hbase.DumpOrderHbaseDAO;
import com.sztx.pay.center.dataaccess.domain.DumpOrderDetailHbaseDO;
import com.sztx.pay.center.dataaccess.domain.DumpOrderHbaseDO;
import com.sztx.pay.center.dataaccess.domain.DumpUserActFlowHbaseDO;
import com.sztx.pay.center.dataaccess.domain.HbaseQueryDO;
import com.sztx.pay.center.dataaccess.entity.QueryOrder;
import com.sztx.se.dataaccess.hbase.impl.BaseHbaseDAOImpl;

@Repository("dumpOrderHbaseDAO")
public class DumpOrderHbaseDAOImpl extends BaseHbaseDAOImpl implements DumpOrderHbaseDAO {

	protected static final Logger log = LoggerFactory.getLogger(DumpOrderHbaseDAOImpl.class);
	
	private static final String TABLE_ORDER = "t_order";

	private static final String TABLE_ORDER_DETAIL = "t_order_detail";

	private static final String TABLE_ACT_FLOW = "t_act_flow";

	@Override
	public boolean addDumpOrder(final DumpOrderHbaseDO hbaseDO) {
		boolean result = hbaseTemplate.execute(TABLE_ORDER, new TableCallback<Boolean>() {
			@Override
			public Boolean doInTable(HTableInterface table) throws Throwable {
				long startTime = System.currentTimeMillis();
				Put put = HbaseUtils.buildPut(hbaseDO, serialize(getOrderRowKeyByTime(hbaseDO.getCreateDate()) + hbaseDO.getTradeOrderId()));
				table.put(put);
				long end = System.currentTimeMillis();
				log.info("本次订单插入hbase耗时:[" + (end -startTime) +"]ms");
				return true;

			}
		});
		return result;
	}

	@Override
	public List<String> findOrderExtend(final String tradeOrderNo,final String qualifier) {
		List<String> result = hbaseTemplate.execute(TABLE_ORDER_DETAIL, new TableCallback<List<String>>() {
			@Override
			public List<String> doInTable(HTableInterface table) throws Throwable {
				List<String> list = new ArrayList<>();
				Get get = new Get(serialize(tradeOrderNo));
				Result result = table.get(get);
				byte[] valueByte = result.getValue(serialize("f_order_detail"), serialize(qualifier));
				String deserialize = deserialize(String.class, valueByte);
				if (qualifier.equals("q_detail")) {
					list.add(deserialize);
				} else {
					list = JSONObject.parseArray(deserialize, String.class);
				}
				return list;
			}
		});
		return result;
	}
	
	@Override
	public boolean batchAddDumpOrder( final List<DumpOrderHbaseDO> hbaseDOs) {
		boolean result = hbaseTemplate.execute(TABLE_ORDER, new TableCallback<Boolean>() {
			@Override
			public Boolean doInTable(HTableInterface table) throws Throwable {
				long startTime = System.currentTimeMillis();
				List<Put> puts = new ArrayList<Put>();
				for (DumpOrderHbaseDO hbaseDO : hbaseDOs) {
					Put put = HbaseUtils.buildPut(hbaseDO, serialize(complexString(String.valueOf(hbaseDO.getTradeOrderType()), 3) + getOrderRowKeyByTime(hbaseDO.getCreateDate()) + hbaseDO.getTradeOrderId()));
					puts.add(put);
				}
				table.put(puts);
				long end = System.currentTimeMillis();
				log.info("本次订单插入hbase耗时:[" + (end -startTime) +"]ms");
				return true;

			}
		});
		return result;
	}
	
	@Override
	public List<String> scanDumpOrder(final String startTime, final String endTime, final Integer orderType) {
		List<String> result = hbaseTemplate.execute(TABLE_ORDER, new TableCallback<List<String>>() {
			@Override
			public List<String> doInTable(HTableInterface table) throws Throwable {
				List<String> list = new ArrayList<String>();
				Scan scan = new Scan();
				if (!StringUtil.isEmpty(endTime)) {
					Long startRowkey = getRowKeyByDay(endTime + " 23:59:59");
					scan.setStartRow(serialize(startRowkey));
				}
				if (!StringUtil.isEmpty(startTime)) {
					Long endRowkey = getRowKeyByDay(startTime + " 00:00:00");
					scan.setStopRow(serialize(endRowkey));
				}
				Filter filter = new SingleColumnValueFilter(serialize("f_order"), serialize("tradeOrderType"), CompareOp.EQUAL, serialize(orderType));
				scan.setFilter(filter);
				ResultScanner rs = table.getScanner(scan);
				for (Result r : rs) {
					System.out.println("获得到rowkey:" + new String(r.getRow()));
					byte[] value = r.getValue(serialize("f_order"), serialize("data"));
					String deserialize = deserialize(String.class, value);
					list.add(deserialize);
				}
				return list;
			}
		});
		return result;
	}

	protected Long getRowKeyByDay(String startTime) {
		return DateUtil.parse("2222-12-31 23:59:59", DateUtil.LONG_DATE_FORMAT_STR).getTime() - DateUtil.parse(startTime, DateUtil.LONG_DATE_FORMAT_STR).getTime();
	}

	@Override
	public boolean addDumpUserActFlow(final List<DumpUserActFlowHbaseDO> hbaseDOs) {
		boolean result = hbaseTemplate.execute(TABLE_ACT_FLOW, new TableCallback<Boolean>() {
			@Override
			public Boolean doInTable(HTableInterface table) throws Throwable {
				long startTime = System.currentTimeMillis();
				List<Put> puts = new ArrayList<Put>();
				for (DumpUserActFlowHbaseDO hbaseDO : hbaseDOs) {
					Long max = 10000000000L;
					Integer modifyCount = hbaseDO.getModifyCount();
					Long temp = max - modifyCount;
					//userId是必传字段,故作为rowKey高位(前12位)的固定位
					Put put = HbaseUtils.buildPut(hbaseDO, serialize(complexString(hbaseDO.getUserId(), 12) + getOrderRowKeyByTime(hbaseDO.getCreateDate()) + "" + temp+ hbaseDO.getActFlowNo()));
					puts.add(put);
				}
				table.put(puts);
				long end = System.currentTimeMillis();
				log.info("本次流水插入hbase耗时:[" + (end -startTime) +"]ms");
				return true;

			}
		});
		return result;
	}

	@Override
	public List<String> scanActFlow(final String startTime, final String endTime, final  String userId) {
		List<String> result = hbaseTemplate.execute(TABLE_ACT_FLOW, new TableCallback<List<String>>() {
			@Override
			public List<String> doInTable(HTableInterface table) throws Throwable {
				List<String> list = new ArrayList<String>();
				Scan scan = new Scan();
				if (!StringUtil.isEmpty(endTime)) {
					Long startRowkey = getRowKeyByDay(endTime + " 23:59:59");
					scan.setStartRow(serialize(startRowkey));
				}
				if (!StringUtil.isEmpty(startTime)) {
					Long endRowkey = getRowKeyByDay(startTime + " 00:00:00");
					scan.setStopRow(serialize(endRowkey));
				}
				Filter filter = new SingleColumnValueFilter(serialize("f_flow"), serialize("userId"), CompareOp.EQUAL, serialize(userId));
				scan.setFilter(filter);
				ResultScanner rs = table.getScanner(scan);
				for (Result r : rs) {
					System.out.println("获得到rowkey:" + new String(r.getRow()));
					byte[] value = r.getValue(serialize("f_flow"), serialize("data"));
					String deserialize = deserialize(String.class, value);
					list.add(deserialize);
				}
				return list;
			}
		});
		return result;
	}
	
	@Override
	public HbaseQueryDO scanCmsActFlow(final String pageStartRowKey, final String originOrderNo,final String startDate,final String endDate,final String userId,final Integer userType,final String businessType,final Integer pageSize) {
		HbaseQueryDO result = hbaseTemplate.execute(TABLE_ACT_FLOW, new TableCallback<HbaseQueryDO>() {
			@Override
			public HbaseQueryDO doInTable(HTableInterface table) throws Throwable {
				long startTime = System.currentTimeMillis();
			
				HbaseQueryDO queryDO = new HbaseQueryDO();
				
				List<String> list = new ArrayList<String>();
				
				Scan scan = new Scan();
				//userId是必传字段,故作为rowKey高位(前12位)的固定位
				String startRowkey = complexString(userId, 12);
				String endRowkey = complexString(userId, 12);
				
				if (!StringUtil.isEmpty(endDate)) {
					startRowkey = startRowkey + getRowKeyByDay(endDate);
				}
				
				if (!StringUtil.isEmpty(pageStartRowKey)) {
					startRowkey = pageStartRowKey;
				}
				
				if (!StringUtil.isEmpty(startDate)) {
					endRowkey = endRowkey + getRowKeyByDay(startDate);
				}
				
				scan.setStartRow(serialize(startRowkey));
				scan.setStopRow(serialize(endRowkey));
				
				List<Filter> listFilters = new ArrayList<Filter>();
				
				PageFilter pageFilter = new PageFilter(pageSize + 1);
//				Filter filter = new SingleColumnValueFilter(serialize("f_flow"), serialize("userId"), CompareOp.EQUAL, serialize(userId));
//				listFilters.add(filter);
				Filter filter1 = new SingleColumnValueFilter(serialize("f_flow"), serialize("actType"), CompareOp.EQUAL, serialize(userType));
				listFilters.add(filter1);
				if (StringUtils.isNotBlank(originOrderNo)) {
					Filter filter2 = new SingleColumnValueFilter(serialize("f_flow"), serialize("originOrderNo"), CompareOp.EQUAL, serialize(originOrderNo));
					listFilters.add(filter2);
				}
				if (StringUtils.isNotBlank(businessType)) {
					Filter filter2 = new SingleColumnValueFilter(serialize("f_flow"), serialize("bussinessType"), CompareOp.EQUAL, serialize(businessType));
					listFilters.add(filter2);
				}
				listFilters.add(pageFilter);
				FilterList filterList = new FilterList(listFilters);
				scan.setFilter(filterList);
				
				ResultScanner rs = table.getScanner(scan);
				int tempNext = -1;
				int tempPre = -1;
				boolean hasNext = false;
				boolean hasPre = false;
				String nextRowKey = null;
				String preRowKey = null;
				long startRenderTime = System.currentTimeMillis();
				for (Result r : rs) {
					System.out.println("获得到rowkey:" + new String(r.getRow()));
					tempNext++;
					if (tempNext == pageSize) {
						hasNext = true;
						nextRowKey = new String(r.getRow());
						break;
					} else {
						byte[] value = r.getValue(serialize("f_flow"), serialize("data"));
						String deserialize = deserialize(String.class, value);
						list.add(deserialize);
					}
				}
				long endRenderTime = System.currentTimeMillis();
				log.info("正扫组装耗时:" + (endRenderTime - startRenderTime));
				long endTime = System.currentTimeMillis();
				log.info("正扫耗时:" + (endTime - startTime));
				queryDO.setNextPageStratRowKey(nextRowKey);
				queryDO.setHasNextPage(hasNext);
				queryDO.setJsonDataList(list);
//				if (!StringUtil.isEmpty(pageStartRowKey)) {
//					long startReverseTime = System.currentTimeMillis();
//					Scan reverseScan = new Scan();
//					reverseScan.setStartRow(serialize(pageStartRowKey));
//					if (!StringUtil.isEmpty(endDate)) {
//						Long startRowkey = getRowKeyByDay(endDate);
//						reverseScan.setStopRow(serialize(startRowkey));
//					}
//					reverseScan.setReversed(true);
//					reverseScan.setFilter(filterList);
//					ResultScanner reverseRs = table.getScanner(reverseScan);
//					
//					long startReverseRenderTime = System.currentTimeMillis();
//					for (Result r : reverseRs) {
//						tempPre++;
//						if (tempPre == pageSize) {
//							hasPre = true;
//							preRowKey = new String(r.getRow());
//							break;
//						}
//					}
//					long endReverseRenderTime = System.currentTimeMillis();
//					log.info("反扫组装耗时:" + (endReverseRenderTime - startReverseRenderTime));
//					long endReverseTime = System.currentTimeMillis();
//					log.info("反扫耗时:" + (startReverseTime - endReverseTime));
//				}
				queryDO.setPrePageStratRowKey(preRowKey);
				queryDO.setHasPrePage(hasPre);
				return queryDO;
			}
		});
		return result;
	}
	
	@Override
	public boolean addOrderDetail(final String tradeOrderId, final String jsonData, final String cloumName) {
		boolean result = hbaseTemplate.execute(TABLE_ORDER_DETAIL, new TableCallback<Boolean>() {

			@Override
			public Boolean doInTable(HTableInterface table) throws Throwable {
				Put put = new Put(serialize(tradeOrderId));
				put.add(serialize("f_order_detail"), serialize(cloumName), serialize(jsonData));
				table.put(put);
				return true;
			}
		});
		return result;
	}

	private static Long getOrderRowKeyByTime(Date date) {
		return DateUtil.parse("2222-12-31 23:59:59", DateUtil.LONG_DATE_FORMAT_STR).getTime() - date.getTime();
	}

	@Override
	public boolean batchAddOrderDetail(final List<DumpOrderDetailHbaseDO> hbaseDOs) {
		boolean result = hbaseTemplate.execute(TABLE_ORDER_DETAIL, new TableCallback<Boolean>() {
			@Override
			public Boolean doInTable(HTableInterface table) throws Throwable {
				long startTime = System.currentTimeMillis();
				List<Put> puts = new ArrayList<Put>();
				for (DumpOrderDetailHbaseDO hbaseDO : hbaseDOs) {
					Put put = HbaseUtils.buildPut(hbaseDO, serialize(hbaseDO.getTradeOrderId()));
					puts.add(put);
				}
				table.put(puts);
				long end = System.currentTimeMillis();
				log.info("本次订单详情插入hbase耗时:[" + (end -startTime) +"]ms");
				return true;

			}
		});
		return result;
	}

	@Override
	public HbaseQueryDO scanActFlow(final String pageStartRowKey,final String originOrderNo,final String startDate,final String endDate,final String userId,final Integer userType,final String startBusinessType,final Integer pageSize,
			final Integer tradeType, final Integer inOrOut, final String endBusinessType) {
		HbaseQueryDO result = hbaseTemplate.execute(TABLE_ACT_FLOW, new TableCallback<HbaseQueryDO>() {
			@Override
			public HbaseQueryDO doInTable(HTableInterface table) throws Throwable {
				long startTime = System.currentTimeMillis();
				HbaseQueryDO queryDO = new HbaseQueryDO();
				
				List<String> list = new ArrayList<String>();
				
				Scan scan = new Scan();
				//userId是必传字段,故作为rowKey高位(前12位)的固定位
				String startRowkey = complexString(userId, 12);
				String endRowkey = complexString(userId, 12);
				
				if (!StringUtil.isEmpty(endDate)) {
					startRowkey = startRowkey + getRowKeyByDay(endDate);
				}
				
				if (!StringUtil.isEmpty(pageStartRowKey)) {
					startRowkey = pageStartRowKey;
				}
				
				if (!StringUtil.isEmpty(startDate)) {
					endRowkey = endRowkey + getRowKeyByDay(startDate);
				}
				
				scan.setStartRow(serialize(startRowkey));
				scan.setStopRow(serialize(endRowkey));
				
				List<Filter> listFilters = new ArrayList<Filter>();
				
				PageFilter pageFilter = new PageFilter(pageSize + 1);
				
//				Filter filter = new SingleColumnValueFilter(serialize("f_flow"), serialize("userId"), CompareOp.EQUAL, serialize(userId));
//				listFilters.add(filter);
				Filter filter1 = new SingleColumnValueFilter(serialize("f_flow"), serialize("actType"), CompareOp.EQUAL, serialize(userType));
				listFilters.add(filter1);
				if (StringUtils.isNotBlank(originOrderNo)) {
					Filter filter2 = new SingleColumnValueFilter(serialize("f_flow"), serialize("originOrderNo"), CompareOp.EQUAL, serialize(originOrderNo));
					listFilters.add(filter2);
				}
				if (StringUtils.isNotBlank(startBusinessType)) {
					Filter filter2 = new SingleColumnValueFilter(serialize("f_flow"), serialize("bussinessType"), CompareOp.GREATER_OR_EQUAL, serialize(startBusinessType));
					listFilters.add(filter2);
				}
				if (StringUtils.isNotBlank(endBusinessType)) {
					Filter filter2 = new SingleColumnValueFilter(serialize("f_flow"), serialize("bussinessType"), CompareOp.LESS_OR_EQUAL, serialize(endBusinessType));
					listFilters.add(filter2);
				}
				if (tradeType != null && tradeType != 0){
					Filter filter2 = new SingleColumnValueFilter(serialize("f_flow"), serialize("tradeType"), CompareOp.EQUAL, serialize(tradeType));
					listFilters.add(filter2);
				}
				if (inOrOut != null){
					if (inOrOut == 1) {
						Filter filter2 = new SingleColumnValueFilter(serialize("f_flow"), serialize("tradeAmount"), CompareOp.GREATER, serialize(0L));
						listFilters.add(filter2);
					} else if (inOrOut == 2) {
						Filter filter2 = new SingleColumnValueFilter(serialize("f_flow"), serialize("tradeAmount"), CompareOp.LESS, serialize(0L));
						listFilters.add(filter2);
					}
				}
				listFilters.add(pageFilter);
				FilterList filterList = new FilterList(listFilters);
				scan.setFilter(filterList);
				
				ResultScanner rs = table.getScanner(scan);
				int tempNext = -1;
				int tempPre = -1;
				boolean hasNext = false;
				boolean hasPre = false;
				String nextRowKey = null;
				String preRowKey = null;
				long startRenderTime = System.currentTimeMillis();
				for (Result r : rs) {
					tempNext++;
					if (tempNext == pageSize) {
						hasNext = true;
						nextRowKey = new String(r.getRow());
						break;
					} else {
						byte[] value = r.getValue(serialize("f_flow"), serialize("data"));
						String deserialize = deserialize(String.class, value);
						list.add(deserialize);
					}
				}
				long endRenderTime = System.currentTimeMillis();
				log.info("正扫组装耗时:" + (endRenderTime - startRenderTime));
				long endTime = System.currentTimeMillis();
				log.info("正扫耗时:" + (endTime - startTime));
				queryDO.setNextPageStratRowKey(nextRowKey);
				queryDO.setHasNextPage(hasNext);
				queryDO.setJsonDataList(list);
//				if (!StringUtil.isEmpty(pageStartRowKey)) {
//					long startReverseTime = System.currentTimeMillis();
//					Scan reverseScan = new Scan();
//					reverseScan.setStartRow(serialize(pageStartRowKey));
//					if (!StringUtil.isEmpty(endDate)) {
//						Long startRowkey = getRowKeyByDay(endDate);
//						reverseScan.setStopRow(serialize(startRowkey));
//					}
//					reverseScan.setReversed(true);
//					reverseScan.setFilter(filterList);
//					ResultScanner reverseRs = table.getScanner(reverseScan);
//					
//					long startReverseRenderTime = System.currentTimeMillis();
//					for (Result r : reverseRs) {
//						tempPre++;
//						if (tempPre == pageSize) {
//							hasPre = true;
//							preRowKey = new String(r.getRow());
//							break;
//						}
//					}
//					long endReverseRenderTime = System.currentTimeMillis();
//					log.info("反扫组装耗时:" + (endReverseRenderTime - startReverseRenderTime));
//					long endReverseTime = System.currentTimeMillis();
//					log.info("反扫耗时:" + (endReverseTime - startReverseTime));
//				}
				queryDO.setPrePageStratRowKey(preRowKey);
				queryDO.setHasPrePage(hasPre);
				return queryDO;
			}
		});
		return result;
	}
	
	public static void main(String[] args) {
//		System.out.println(DateUtil.parse("2030-12-31 23:59:59", DateUtil.LONG_DATE_FORMAT_STR).getTime() - DateUtil.parse("2016-08-30 11:19:23", DateUtil.LONG_DATE_FORMAT_STR).getTime());
//		System.out.println(DateUtil.parse("2030-12-31 23:59:59", DateUtil.LONG_DATE_FORMAT_STR).getTime() - DateUtil.parse("2016-12-30 23:59:59", DateUtil.LONG_DATE_FORMAT_STR).getTime());
//		String l1 = "92233705618072868071820160928001040";
//		String l2 = "92233705618072868071820160928001042";
		getTradeOrderRowKey(new Date(), "555", 1);
	}

	@Override
	public HbaseQueryDO scanDumpOrder(final QueryOrder queryOrder,final String pageStartRowKey,final Integer pageSize) {
		HbaseQueryDO result = hbaseTemplate.execute(TABLE_ORDER, new TableCallback<HbaseQueryDO>() {
			@Override
			public HbaseQueryDO doInTable(HTableInterface table) throws Throwable {
				
				String startDate = queryOrder.getStartTime();
				String endDate = queryOrder.getEndTime();
				Integer channelType = queryOrder.getChannelType();
				String orderStatus = queryOrder.getOrderStatus();
				String payeeAccountId = queryOrder.getPayeeAccountId();
				String payerAccountId = queryOrder.getPayerAccountId();
				String payOrderId = queryOrder.getPayOrderId();
				String tradeOrderId = queryOrder.getTradeOrderId();
				Integer tradeOrderType = queryOrder.getTradeOrderType();
				String businessType = queryOrder.getBusinessType();
				HbaseQueryDO queryDO = new HbaseQueryDO();
				
				List<String> list = new ArrayList<String>();
				
				Scan scan = new Scan();
				
				//订单类型是必传条件,故作为rowKey高位(前三位)固定位.
				String startRowkey = complexString(String.valueOf(tradeOrderType), 3);
				String endRowkey = complexString(String.valueOf(tradeOrderType), 3);
				
				if (!StringUtil.isEmpty(endDate)) {
					startRowkey = startRowkey + getRowKeyByDay(endDate);
				}
				
				if (!StringUtil.isEmpty(pageStartRowKey)) {
					startRowkey = pageStartRowKey;
					
				}
				if (!StringUtil.isEmpty(startDate)) {
					endRowkey = endRowkey + getRowKeyByDay(startDate);
				}
				
				scan.setStartRow(serialize(startRowkey));
				scan.setStopRow(serialize(endRowkey));
				
				List<Filter> listFilters = new ArrayList<Filter>();
				
				PageFilter pageFilter = new PageFilter(pageSize + 1);
				
//				Filter filter = new SingleColumnValueFilter(serialize("f_order"), serialize("tradeOrderType"), CompareOp.EQUAL, serialize(tradeOrderType));
//				listFilters.add(filter);
				if (StringUtils.isNotBlank(orderStatus)) {
					Filter filter1 = new SingleColumnValueFilter(serialize("f_order"), serialize("orderStatus"), CompareOp.EQUAL, serialize(new Integer(orderStatus)));
					listFilters.add(filter1);
				}
				if (StringUtils.isNotBlank(payeeAccountId)) {
					Filter filter1 = new SingleColumnValueFilter(serialize("f_order"), serialize("payeeAccountId"), CompareOp.EQUAL, serialize(payeeAccountId));
					listFilters.add(filter1);
				}
				if (StringUtils.isNotBlank(payerAccountId)) {
					Filter filter1 = new SingleColumnValueFilter(serialize("f_order"), serialize("payerAccountId"), CompareOp.EQUAL, serialize(payerAccountId));
					listFilters.add(filter1);
				}
				if (StringUtils.isNotBlank(payOrderId)) {
					Filter filter1 = new SingleColumnValueFilter(serialize("f_order"), serialize("payOrderId"), CompareOp.EQUAL, serialize(payOrderId));
					listFilters.add(filter1);
				}
				if (StringUtils.isNotBlank(tradeOrderId)) {
					Filter filter1 = new SingleColumnValueFilter(serialize("f_order"), serialize("tradeOrderId"), CompareOp.EQUAL, serialize(tradeOrderId));
					listFilters.add(filter1);
				}
				if (StringUtils.isNotBlank(businessType)) {
					Filter filter1 = new SingleColumnValueFilter(serialize("f_order"), serialize("bussinessType"), CompareOp.EQUAL, serialize(businessType));
					listFilters.add(filter1);
				}
				if (channelType != null) {
					if (channelType == 1) {
						Filter filter1 = new SingleColumnValueFilter(serialize("f_order"), serialize("channelId"), CompareOp.EQUAL, serialize("99"));
						listFilters.add(filter1);
					} else if (channelType == 2) {
						Filter filter1 = new SingleColumnValueFilter(serialize("f_order"), serialize("channelId"), CompareOp.NOT_EQUAL, serialize("99"));
						listFilters.add(filter1);
					}
				}
				listFilters.add(pageFilter);
				FilterList filterList = new FilterList(listFilters);
				scan.setFilter(filterList);
				
				ResultScanner rs = table.getScanner(scan);
				int tempNext = -1;
				int tempPre = -1;
				boolean hasNext = false;
				boolean hasPre = false;
				String nextRowKey = null;
				String preRowKey = null;
				for (Result r : rs) {
					System.out.println("获得到rowkey:" + new String(r.getRow()));
					tempNext++;
					if (tempNext == pageSize) {
						hasNext = true;
						nextRowKey = new String(r.getRow());
						break;
					} else {
						byte[] value = r.getValue(serialize("f_order"), serialize("data"));
						String deserialize = deserialize(String.class, value);
						list.add(deserialize);
					}
				}
				queryDO.setNextPageStratRowKey(nextRowKey);
				queryDO.setHasNextPage(hasNext);
				queryDO.setJsonDataList(list);
//				if (!StringUtil.isEmpty(pageStartRowKey)) {
//					Scan reverseScan = new Scan();
//					reverseScan.setStartRow(serialize(pageStartRowKey));
//					if (!StringUtil.isEmpty(endDate)) {
//						Long startRowkey = getRowKeyByDay(endDate);
//						reverseScan.setStopRow(serialize(startRowkey));
//					}
//					reverseScan.setReversed(true);
//					reverseScan.setFilter(filterList);
//					ResultScanner reverseRs = table.getScanner(reverseScan);
//					
//					for (Result r : reverseRs) {
//						tempPre++;
//						if (tempPre == pageSize) {
//							hasPre = true;
//							preRowKey = new String(r.getRow());
//							break;
//						}
//					}
//				}
				queryDO.setPrePageStratRowKey(preRowKey);
				queryDO.setHasPrePage(hasPre);
				return queryDO;
			}
		});
		return result;
	}
	
	private String complexString(String data, Integer length) {
		
		int dataLength = data.length();
		
		for (int i = 0; i < (length - dataLength); i++) {
			data = "0" + data;
		}
		
		return data;
	}
	
	private static String getTradeOrderRowKey(Date createDate, String tradeOrderId, Integer tradeOrderType){
		
		String orderTypeString  = String.valueOf(tradeOrderType);
		int length = orderTypeString.length();
		for (int i = 0; i < (3-length); i++) {
			orderTypeString = "0" + orderTypeString;
		}
		StringBuffer rowKey = new StringBuffer();
		rowKey.append(orderTypeString).append(getOrderRowKeyByTime(createDate)).append(tradeOrderId);
		
		return rowKey.toString();
	}
	
}
